package com.hanson.demo.queue.memory;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.springframework.stereotype.Component;

import com.hanson.demo.actors.Message;
import com.hanson.demo.queue.TbQueueConsumer;
import com.hanson.demo.service.UserActorService;

import cn.hutool.json.JSONUtil;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
@RequiredArgsConstructor
public class InMemoryTbQueueConsumer<T extends Message> implements TbQueueConsumer<T> {
	
    private volatile boolean stopped =false;
    private final InMemoryStorage storage = InMemoryStorage.getInstance();
    private final UserActorService userActorService;
    
    protected volatile ExecutorService consumersExecutor= Executors.newFixedThreadPool(1); // 并发任务; 

    
    @PostConstruct
    public void startMemoryConsumer() {
    	consumersExecutor.submit(() -> {
			while (true) {
				try {
					Map<String,Message> msgMap = storage.poll(); // 移除并返回队列头部的元素 如果队列为空，则阻塞
					if(msgMap!=null) {
						for (String key : msgMap.keySet()) {
							this.consumerMessage(key,JSONUtil.toJsonStr(msgMap.get(key)));
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
					Thread.currentThread().interrupt();
				}
			}
		});
    }
  
    @Override
    public boolean isStopped() {
        return stopped;
    }

	@Override
	public void consumerMessage(String topic, String msg) {
		log.info("==>consumerMessage:topic{},msg:{}", topic, msg);
		userActorService.dealQueueConsumer(topic,msg);
	}

}
